/*
 * Title:        CloudScheduler Toolkit
 * Description:  a Toolkit for Modeling and Simulation of Job Scheduling and Resource Provisioning in Cloud System
 * Licence:      GPL - http://www.gnu.org/copyleft/gpl.html
 * Author:       Fuhui Wu
 *
 * Copyright (c) 2013-2014, The National University of Defense Technology, China
 */

package org.nudt.jCloud.example.resourceStealing2;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.nudt.jCloud.infrastructure.Cluster;
import org.nudt.jCloud.infrastructure.Slot;
import org.nudt.jCloud.infrastructure.SlotType;
import org.nudt.jCloud.simulation.CloudSchedSIM;
import org.nudt.jCloud.simulation.Framework;
import org.nudt.jCloud.simulation.RequestData;
import org.nudt.jCloud.simulation.ScaleData;
import org.nudt.jCloud.simulation.UserBroker;

public class AvgElasticHeuristicPreemptionRM extends StealingRM{
	private double upThreshold;
	private double downThreshold;

	public AvgElasticHeuristicPreemptionRM(String name, Cluster cluster,
			double upThreshold, double downThreshold) {
		super(name, cluster);
		this.upThreshold = upThreshold;
		this.downThreshold = downThreshold;
	}
	
	public void jointLease(List<Framework> requestFrwkList, List<ScaleData> scaleDataList){
		Map<Framework, Integer> loadDistr = new HashMap<Framework, Integer>();
		Map<SlotType, Integer> requestLoad = new HashMap<SlotType, Integer>();
		for(Framework frwk : getFrwkList()){
			if(requestFrwkList.contains(frwk)==false){
				RequestData requestData = new RequestData();
				requestData.init(frwk.getJobScheduler().request(),
						frwk.getUserBroker().getRM_Mode());
				Map<SlotType, Integer> curLoad = requestData.requestLoad();
				int unAssignedLoad = 0;
				for(SlotType slotType : curLoad.keySet()){
					unAssignedLoad += curLoad.get(slotType);
					/*if(requestLoad.containsKey(slotType)){
						requestLoad.put(slotType, requestLoad.get(slotType) + curLoad.get(slotType));
					}else{
						requestLoad.put(slotType, curLoad.get(slotType));
					}*/
				}
				loadDistr.put(frwk, unAssignedLoad+frwk.busySlotNum());
			}
		}
		for(ScaleData scaleData : scaleDataList){
			int userBrokerId = scaleData.getUserBrokerId();
			UserBroker userBroker = (UserBroker)CloudSchedSIM.getEntityById(userBrokerId);
			Map<SlotType, Integer> curLoad = scaleData.getRequestData().requestLoad();
			int unAssignedLoad = 0;
			for(SlotType slotType : curLoad.keySet()){
				unAssignedLoad += curLoad.get(slotType);
				if(requestLoad.containsKey(slotType)){
					requestLoad.put(slotType, requestLoad.get(slotType) + curLoad.get(slotType)
							+userBroker.getFrwk().busySlotNum());
				}else{
					requestLoad.put(slotType, curLoad.get(slotType)+userBroker.getFrwk().busySlotNum());
				}
			}			
			loadDistr.put(userBroker.getFrwk(), unAssignedLoad+userBroker.getFrwk().busySlotNum());
		}
		int availableNum = 0;
		for(Framework frwk : getFrwkList()){
			if(requestFrwkList.contains(frwk)){
				availableNum += frwk.busySlotNum()+frwk.freeSlotNum();
			}else{
				availableNum += frwk.freeSlotNum();
			}
		}
		int detaNum = 0;	
		boolean leasing = false;
		for(SlotType slotType : requestLoad.keySet()){
			if(availableNum == 0){
				detaNum = requestLoad.get(slotType);
				leasing = true;
			}else{
				double curLoad = ((double)(requestLoad.get(slotType)))/availableNum;
				if(curLoad > upThreshold){
					detaNum = (int) Math.ceil((requestLoad.get(slotType))/upThreshold)-availableNum;
					leasing = true;
				}else if(curLoad < downThreshold){
					detaNum = availableNum - (int) Math.ceil((requestLoad.get(slotType))/downThreshold);
					while(detaNum>0){
						Framework frwk = mostOcpWithFreeNoReq(requestFrwkList, getFrwkList());
						if(frwk == null){
							break;
						}else{
							List<Slot> releasedSlotList = frwk.releaseFreeSlot(slotType, 1);
							if(releasedSlotList.size()>0){
								releaseSlot(releasedSlotList);
								detaNum = detaNum-1;
							}else{
								break;
							}			
						}
					}
					while(detaNum>0){
						Framework frwk = mostOcpwithFreeInReq(requestFrwkList);
						if(frwk == null){
							break;
						}else{
							List<Slot> releasedSlotList = frwk.releaseFreeSlot(slotType, 1);
							if(releasedSlotList.size()>0){
								releaseSlot(releasedSlotList);
								detaNum = detaNum-1;
							}else{
								break;
							}							
						}
					}
				}
			}
		}
		if(leasing == true){
			boolean noAvailableRes = false;
			while(detaNum>0){
				ScaleData scaleData = leastOcpInReq(scaleDataList);
				if(scaleData == null){
					break;
				}else{
					for(SlotType slotType : scaleData.getToLeaseSlot().keySet()){
						List<Slot> leasedSlotList = provideSlots(slotType, 1);
						if(leasedSlotList.size()>0){
							UserBroker userBroker = (UserBroker)CloudSchedSIM.getEntityById(scaleData.getUserBrokerId());
							Framework frwk = userBroker.getFrwk();
							frwk.leaseSlot(slotType, leasedSlotList);
							detaNum = detaNum-1;
						}else{
							noAvailableRes = true;
							break;
						}
					}
					if(noAvailableRes){
						break;
					}
				}
			}			
		}
		/*double totalLoad = 0;
		for(Framework frwk : loadDistr.keySet()){
			totalLoad += loadDistr.get(frwk);
		}
		for(Framework frwk : getFrwkList()){
			if(loadDistr.get(frwk)==0){
				getReservationRatio().put(frwk, Double.MIN_VALUE);
			}else{
				getReservationRatio().put(frwk, loadDistr.get(frwk)/totalLoad);				
			}
		}*/
	}
	
	@Override
	public void scaleResource(List<ScaleData> scaleDataList) {
		if(isInitialized() == false){
			initSlotRatio();
			setTotalSlotNum(Integer.MAX_VALUE);
		}
		List<Framework > requestFrwkList = requestFrwk(scaleDataList);		
		jointLease(requestFrwkList, scaleDataList);
		resetTotalSlotNum();
		
		stealFromNoReq(requestFrwkList, scaleDataList);
		stealFromReqSatisfied(scaleDataList);
		stealFromReqOverThresh(scaleDataList);	
		
		preempt(requestFrwkList, scaleDataList);	
	}

}
